/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.server;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;

import org.slf4j.Logger;
import org.sss.common.io.StreamDB;
import org.sss.common.io.StreamDBSet;
import org.sss.mapreduce.Decoder;
import org.sss.mapreduce.Encodings;
import org.sss.mapreduce.GroupID;
import org.sss.mapreduce.SideData;
import org.sss.mapreduce.SssException;
import org.sss.mapreduce.StorageNode;
import org.sss.mapreduce.Tuple;
import org.sss.mapreduce.datatype.Packable;

public class SideDataInAdvanceStream<K extends Packable, V extends Packable> implements SideData<K, V> {
  @SuppressWarnings("unused")
  private static Logger logger = SssServer.getLogger(SideDataInAdvanceStream.class);

  private final Map<K, List<V>> dic = new HashMap<K, List<V>>();

  public static <KEY extends Packable, VALUE extends Packable> SideData<KEY, VALUE> create(
      SssServerResource sssResouce,
      StorageNode node,
      GroupID gid,
      Encodings encodings,
      Class<KEY> keyClass,
      Class<VALUE> valueClass) throws SssException, IOException, ExecutionException, InterruptedException {
    return new SideDataInAdvanceStream<KEY, VALUE>(sssResouce, node, gid, encodings, keyClass, valueClass);
  }

  // TODO: uses appropriate exception.
  // TODO: issue database requests from i/o thread.
  private SideDataInAdvanceStream(SssServerResource sssResouce,
                  StorageNode nearNode,
                  GroupID gid,
                  Encodings encodings,
                  Class<K> keyClass,
                  Class<V> valueClass) throws SssException, IOException, ExecutionException, InterruptedException {
    StreamDBSet dbSet = sssResouce.getStreamDBSetForSideData();
    try {
      Decoder keyDec = encodings.createDecoder(keyClass);
      Decoder valDec = encodings.createDecoder(valueClass);

      if (gid.isBroadcast()) {
        read(dbSet.get(nearNode), gid, keyClass, valueClass, keyDec, valDec, dic);
      }
      else {
        for (StorageNode node: sssResouce.getStorageNodeList()) {
            read(dbSet.get(node), gid, keyClass, valueClass, keyDec, valDec, dic);
        }
      }
    }
    finally {
      dbSet.close();
    }
  }

  public boolean containsKey(K key) {
    return dic.containsKey(key);
  }

  // TODO: uses appropriate exception.
  public synchronized List<V> get(K key) throws SssException {
    return dic.get(key);
  }

  public Iterable<K> keys() throws SssException {
    return dic.keySet();
  }

  public Iterable<Tuple<K, V>> tuples() throws SssException {
    return new Iterable<Tuple<K, V>>() {
      @Override
      public Iterator<Tuple<K, V>> iterator() {
        return new TupleIterator<K, V>(SideDataInAdvanceStream.this);
      }
    };
  }

  @Override
  public synchronized void close() throws IOException {
    /* Do nothing */
  }

  private static <K extends Packable, V extends Packable> void
    read(StreamDB db,
        GroupID groupid,
        Class<K> keyClass,
        Class<V> valClass,
        Decoder keyDec, Decoder valDec,
        Map<K, List<V>> container) throws SssException, IOException {

    for (int i = 0;i < groupid.getNSlot();++i) {
      Iterator<byte[]> it = db.read(groupid, i, false).iterator();
      while (it.hasNext()) {
        K key   = keyClass.cast(keyDec.decode(it.next()));
        V value = valClass.cast(valDec.decode(it.next()));

        List<V> values = container.get(key);
        if (values == null) {
          values = new ArrayList<V>();
          container.put(key, values);
        }
        values.add(value);
      }
    }
  }
}
